package org.quickserver.net.client.pool;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.quickserver.net.client.BlockingClient;
import org.quickserver.net.client.ClientInfo;
import org.quickserver.net.client.Host;
import org.quickserver.net.client.HostList;
import org.quickserver.net.client.SocketBasedHost;
import org.quickserver.net.client.loaddistribution.LoadDistributor;
import org.quickserver.net.client.loaddistribution.LoadPattern;
import org.quickserver.net.client.loaddistribution.impl.RoundRobinLoadPattern;
import org.quickserver.net.client.monitoring.HostMonitor;
import org.quickserver.net.client.monitoring.HostMonitoringService;
import org.quickserver.net.client.monitoring.HostStateListener;
import org.quickserver.net.client.monitoring.impl.HttpMonitor;




public class BlockingClientPool
{
  private static final Logger logger = Logger.getLogger(BlockingClientPool.class.getName());
  
  private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
  
  private static int maxTimeToLockInSec = 5;
  
  private PoolableBlockingClient poolableBlockingClient;
  
  private String name;
  
  private int minPoolSize;
  
  private int maxPoolSize;
  private int idlePoolSize;
  private Map<SocketBasedHost, ConcurrentLinkedQueue<PooledBlockingClient>> pool = new ConcurrentHashMap<SocketBasedHost, ConcurrentLinkedQueue<PooledBlockingClient>>();

  
  private Map<SocketBasedHost, ConcurrentLinkedQueue<PooledBlockingClient>> inUsePool = new ConcurrentHashMap<SocketBasedHost, ConcurrentLinkedQueue<PooledBlockingClient>>();
  
  private HostMonitoringService hostMonitoringService = new HostMonitoringService();
  
  private Thread noopThread;
  private boolean debug = false;
  private int logPoolStatsTimeInMinute = 0;
  
  private Thread logPoolStats;
  
  public boolean isDebug() { return this.debug; }

  
  public void setDebug(boolean aDebug) { this.debug = aDebug; }

  
  public static void test() {
    HostList hostList = new HostList("myservers");

    
    final HttpMonitor hm = new HttpMonitor();
    final LoadDistributor ld = new LoadDistributor(hostList);
    ld.setLoadPattern((LoadPattern)new RoundRobinLoadPattern());
    
    PoolableBlockingClient poolableBlockingClient = new PoolableBlockingClient()
      {
        public HostMonitor getHostMonitor() { return hm; }


        
        public LoadDistributor getLoadDistributor() { return ld; }


        
        public BlockingClient createBlockingClient(SocketBasedHost host) { throw new UnsupportedOperationException("Not supported yet."); }


        
        public boolean closeBlockingClient(BlockingClient blockingClient) { throw new UnsupportedOperationException("Not supported yet."); }


        
        public boolean sendNoOp(BlockingClient blockingClient) { throw new UnsupportedOperationException("Not supported yet."); }


        
        public long getNoOpTimeIntervalMiliSec() { throw new UnsupportedOperationException("Not supported yet."); }


        
        public int getHostMonitoringIntervalInSec() { throw new UnsupportedOperationException("Not supported yet."); }


        
        public boolean isBlockWhenEmpty() { return false; }


        
        public int getMaxIntervalForBorrowInSec() { return 30; }
      };
  }

  
  public BlockingClientPool(String name, PoolableBlockingClient poolableBlockingClient) {
    this.name = name;
    this.poolableBlockingClient = poolableBlockingClient;
  }
  
  public void init() {
    logger.log(Level.FINEST, "Started with {0}", this.name);
    
    logger.log(Level.FINEST, "HostMonitoringIntervalInSec: {0}", 
        Integer.valueOf(this.poolableBlockingClient.getHostMonitoringIntervalInSec()));
    logger.log(Level.FINEST, "MaxIntervalForBorrowInSec: {0}", 
        Integer.valueOf(this.poolableBlockingClient.getMaxIntervalForBorrowInSec()));
    logger.log(Level.FINEST, "NoOpTimeIntervalMiliSec: {0}", 
        Long.valueOf(this.poolableBlockingClient.getNoOpTimeIntervalMiliSec()));
    logger.log(Level.FINEST, "MinPoolSize: {0}", Integer.valueOf(getMinPoolSize()));
    logger.log(Level.FINEST, "IdlePoolSize: {0}", Integer.valueOf(getIdlePoolSize()));
    logger.log(Level.FINEST, "MaxPoolSize: {0}", Integer.valueOf(getMaxPoolSize()));

    
    LoadDistributor ld = getPoolableBlockingClient().getLoadDistributor();
    if (ld == null) throw new NullPointerException("Load Distributor is not set!");
    
    HostMonitor hm = getPoolableBlockingClient().getHostMonitor();
    if (hm == null) throw new NullPointerException("Host Monitor is not set!");

    
    getHostMonitoringService().setHostList(ld.getHostList());
    getHostMonitoringService().setHostMonitor(hm);
    getHostMonitoringService().setIntervalInSec(
        getPoolableBlockingClient().getHostMonitoringIntervalInSec());
    
    HostStateListener hsl = new HostStateListener() {
        public void stateChanged(Host host, char oldstatus, char newstatus) {
          if (oldstatus != 'U') {
            logger.log(Level.WARNING, "State changed: {0}; old state: {1};new state: {2}", 
                new Object[] { host, Character.valueOf(oldstatus), Character.valueOf(newstatus) });
            
            SocketBasedHost shost = (SocketBasedHost)host;
            ConcurrentLinkedQueue poolForHost = (ConcurrentLinkedQueue)BlockingClientPool.this.pool.get(shost);
            if (newstatus == 'A') {
              BlockingClientPool.this.increaseSize(shost, poolForHost);
            } else {
              BlockingClientPool.this.cleanPool(shost);
            } 
          } else {
            logger.log(Level.INFO, "State changed: {0}; old state: {1};new state: {2}", 
                new Object[] { host, Character.valueOf(oldstatus), Character.valueOf(newstatus) });
          } 
        }
      };
    getHostMonitoringService().addHostStateListner(hsl);
    HostMonitoringService.add(getHostMonitoringService());
    HostMonitoringService.monitor(true, getHostMonitoringService());
    
    HostList hostlist = ld.getHostList();
    
    List fullHostList = hostlist.getFullList();
    Iterator<SocketBasedHost> iterator = fullHostList.iterator();
    
    SocketBasedHost host = null;
    ConcurrentLinkedQueue<PooledBlockingClient> poolForHost = null;
    ConcurrentLinkedQueue<PooledBlockingClient> poolForInUseHost = null;
    
    if (this.noopThread != null) {
      this.noopThread.interrupt();
      this.noopThread = null;
    } 


    
    this.lock.writeLock().lock();
    try {
      while (iterator.hasNext()) {
        host = iterator.next();
        poolForHost = this.pool.get(host);
        if (poolForHost != null) {
          cleanPool(host);
        } else {
          poolForHost = new ConcurrentLinkedQueue();
          this.pool.put(host, poolForHost);
          
          poolForInUseHost = new ConcurrentLinkedQueue();
          this.inUsePool.put(host, poolForInUseHost);
        } 
        int _poolSize = 0;
        
        if (host.getStatus() == 'A') {
          PooledBlockingClient pooledBlockingClient = null;
          while (_poolSize++ < getMinPoolSize()) {
            pooledBlockingClient = getNewPooledBlockingClient(host);
            if (pooledBlockingClient == null) {
              _poolSize--;
              break;
            } 
            poolForHost.add(pooledBlockingClient);
          } 
        } 
      } 
    } finally {
      this.lock.writeLock().unlock();
    } 
    
    this.noopThread = new Thread() {
        public void run() {
          long stime = 0L;
          long timeTaken = 0L;
          
          long timeToSlepp = 0L;
          while (true) {
            timeToSlepp = BlockingClientPool.this.poolableBlockingClient.getNoOpTimeIntervalMiliSec() / 2L - timeTaken;
            if (timeToSlepp > 0L) {
              try {
                sleep(timeToSlepp);
              } catch (InterruptedException ex) {
                logger.log(Level.FINEST, "closing noop: {0}", ex);
                
                break;
              } 
            }
            stime = System.currentTimeMillis();
            try {
              BlockingClientPool.this.sendNoOp();
            } catch (Throwable ex) {
              Logger.getLogger(BlockingClientPool.class.getName()).log(Level.SEVERE, 
                  "Error: " + ex, ex);
            } 
            timeTaken = System.currentTimeMillis() - stime;
          } 
        }
      };
    this.noopThread.setName(String.valueOf(this.name) + "-SendNOOP-Thread");
    this.noopThread.setDaemon(true);
    this.noopThread.start();

    
    if (this.logPoolStats != null) {
      this.logPoolStats.interrupt();
      this.logPoolStats = null;
    } 
    if (getLogPoolStatsTimeInMinute() > 0) {
      this.logPoolStats = new Thread() {
          public void run() {
            while (true) {
              try {
                sleep((BlockingClientPool.this.getLogPoolStatsTimeInMinute() * 60 * 1000));
              } catch (InterruptedException ex) {
                Logger.getLogger(BlockingClientPool.class.getName()).log(Level.WARNING, 
                    "Error: " + ex, ex);
                
                break;
              } 
              try {
                logger.log(Level.INFO, "Stats: \r\n{0}", BlockingClientPool.this.getStats());
              } catch (Exception ex) {
                logger.log(Level.WARNING, "Error: " + ex, ex);
              } 
            } 
          }
        };
      this.logPoolStats.setName(String.valueOf(this.name) + "-LogPoolStats-Thread");
      this.logPoolStats.setDaemon(true);
      this.logPoolStats.start();
    } 
    
    logger.log(Level.FINEST, "Done with {0}", this.name);
  }

  
  public PooledBlockingClient getBlockingClient() { return getBlockingClient(null); }


  
  public void checkAllNodes() { HostMonitoringService.monitor(); }






  
  public int getPoolSize(SocketBasedHost host) {
    ConcurrentLinkedQueue poolForHost = this.pool.get(host);
    ConcurrentLinkedQueue poolToHost = this.inUsePool.get(host);
    if (poolForHost == null || poolToHost == null) {
      throw new IllegalStateException("pool for host was null!");
    }
    if (poolForHost.isEmpty() && poolToHost.isEmpty()) return 0;
    
    int size = 0;
    try {
      if (this.lock.readLock().tryLock(maxTimeToLockInSec, TimeUnit.SECONDS)) {
        try {
          size = poolForHost.size() + poolToHost.size();
        } finally {
          this.lock.readLock().unlock();
        } 
      } else {
        return -1;
      } 
    } catch (InterruptedException ex) {
      logger.log(Level.WARNING, "Error: " + ex);
      return -1;
    } 
    return size;
  }
  
  public PooledBlockingClient[] getOneBlockingClientForAllActiveHosts() {
    List<PooledBlockingClient> allClients = new ArrayList<PooledBlockingClient>();
    
    List<SocketBasedHost> listOfActiveHost = this.hostMonitoringService.getHostList().getActiveList();
    Iterator<SocketBasedHost> iterator = listOfActiveHost.iterator();
    
    PooledBlockingClient pooledBlockingClient = null;
    SocketBasedHost host = null;
    while (iterator.hasNext()) {
      host = iterator.next();
      pooledBlockingClient = getBlockingClientByHost(host);
      if (pooledBlockingClient == null) {
        logger.warning("Error getting client from " + host);
        continue;
      } 
      allClients.add(pooledBlockingClient);
    } 
    
    return allClients.toArray(new PooledBlockingClient[0]);
  }
  
  public PooledBlockingClient getBlockingClient(ClientInfo clientInfo) {
    SocketBasedHost host = 
      (SocketBasedHost)getPoolableBlockingClient().getLoadDistributor().getHost(clientInfo);
    if (host == null) {
      logger.log(Level.WARNING, "LoadDistributor.. gave null host!");
      return null;
    } 
    
    if (host.getStatus() != 'A' && host.getStatus() != 'U') {
      logger.log(Level.WARNING, "host is not up! sending null host!");
      return null;
    } 
    
    return getBlockingClientByHost(host);
  }
  
  private PooledBlockingClient getBlockingClientByHost(SocketBasedHost host) {
    ConcurrentLinkedQueue<PooledBlockingClient> poolForHost = this.pool.get(host);
    ConcurrentLinkedQueue<PooledBlockingClient> poolForInUseHost = this.inUsePool.get(host);
    if (poolForHost == null || poolForInUseHost == null) {
      throw new IllegalStateException("pool for host was null!");
    }
    PooledBlockingClient pooledBlockingClient = null;
    
    if (poolForHost.isEmpty()) {
      try {
        int poolsize = poolForHost.size() + poolForInUseHost.size();
        if (poolsize < getMaxPoolSize()) {
          try {
            if (this.lock.readLock().tryLock(maxTimeToLockInSec, TimeUnit.SECONDS)) {
              try {
                poolsize = poolForHost.size() + poolForInUseHost.size();
                if (poolsize < getMaxPoolSize()) {
                  pooledBlockingClient = getNewPooledBlockingClient(host);
                  return pooledBlockingClient;
                } 
              } finally {
                this.lock.readLock().unlock();
              } 
            } else {
              logger.log(Level.WARNING, "not able to get read lock..");
              return null;
            } 
          } catch (InterruptedException ex) {
            logger.log(Level.WARNING, "not able to get read lock..{0}", ex);
            return null;
          } 
        } else if (getPoolableBlockingClient().isBlockWhenEmpty()) {
          for (int i = 0; poolForHost.isEmpty() && i < 10; i++) {
            if (isDebug()) logger.log(Level.FINE, "Socket pool empty.. will wait {0}", Integer.valueOf(i)); 
            synchronized (poolForHost) {
              try {
                poolForHost.wait();
              } catch (InterruptedException e) {
                logger.warning("Interrupted while sleeping" + e);
              } 
              pooledBlockingClient = poolForHost.poll();
            } 
            if (pooledBlockingClient != null)
              break; 
          } 
        }  return pooledBlockingClient;
      } finally {
        if (pooledBlockingClient != null) {
          pooledBlockingClient.setHandedOut(true);
          pooledBlockingClient.setPoolToReturn(poolForHost);
          pooledBlockingClient.setLastActionTime(System.currentTimeMillis());
          
          poolForInUseHost.add(pooledBlockingClient);
        } 
      } 
    }
    this.lock.writeLock().lock();
    try {
      pooledBlockingClient = poolForHost.poll();
      if (pooledBlockingClient != null) {
        pooledBlockingClient.setHandedOut(true);
        pooledBlockingClient.setPoolToReturn(poolForHost);
        pooledBlockingClient.setLastActionTime(System.currentTimeMillis());
        
        poolForInUseHost.add(pooledBlockingClient);
      } 
    } finally {
      this.lock.writeLock().unlock();
    } 
    return pooledBlockingClient;
  }

  
  private PooledBlockingClient getNewPooledBlockingClient(SocketBasedHost host) {
    if (host.getStatus() == 'A') {
      return new PooledBlockingClient(getPoolableBlockingClient(), host);
    }
    int size = getPoolSize(host);
    if (size > 0) {
      logger.log(Level.FINEST, "Host is not UP {0}; size: {1}", new Object[] { host, Integer.valueOf(size) });
      cleanPool(host);
      logger.log(Level.FINEST, "Done {0}; size: {1}", new Object[] { host, Integer.valueOf(getPoolSize(host)) });
    } 
    return null;
  }

  
  public void returnBlockingClient(PooledBlockingClient pooledBlockingClient) {
    if (pooledBlockingClient == null)
      return;  if (pooledBlockingClient.getSocketBasedHost() == null)
      return; 
    ConcurrentLinkedQueue poolForInUseHost = this.inUsePool.get(pooledBlockingClient.getSocketBasedHost());
    
    if (pooledBlockingClient.getBlockingClient() == null) {
      poolForInUseHost.remove(pooledBlockingClient); return;
    } 
    if (!pooledBlockingClient.getBlockingClient().isConnected()) {
      try {
        pooledBlockingClient.getBlockingClient().close();
      } catch (IOException ex) {
        Logger.getLogger(BlockingClientPool.class.getName()).log(Level.WARNING, "Error: " + ex, ex);
      } 
      poolForInUseHost.remove(pooledBlockingClient);
      
      return;
    } 
    if (pooledBlockingClient.getLastActionTime() != 0L) {
      long timepassed = System.currentTimeMillis() - pooledBlockingClient.getLastActionTime();
      
      if (timepassed < 1000L) {
        pooledBlockingClient.returnToPool(poolForInUseHost, this.lock);
        
        return;
      } 
      if (timepassed / 1000L < this.poolableBlockingClient.getHostMonitoringIntervalInSec()) {
        pooledBlockingClient.returnToPool(poolForInUseHost, this.lock);
        
        return;
      } 
    } 
    boolean flag = getPoolableBlockingClient().sendNoOp(
        pooledBlockingClient.getBlockingClient());
    if (isDebug()) {
      logger.log(Level.FINEST, "noop for {0} was {1}", 
          new Object[] { pooledBlockingClient, Boolean.valueOf(flag) });
    }
    if (flag) {
      pooledBlockingClient.returnToPool(poolForInUseHost, this.lock);
    } else {
      poolForInUseHost.remove(pooledBlockingClient);
      
      if (pooledBlockingClient.replaceBlockingClient()) {
        pooledBlockingClient.returnToPool(null, this.lock);
      }
    } 
  }
  
  public String getStats() {
    SocketBasedHost socketBasedHost = null;
    StringBuilder sb = new StringBuilder();
    String stat = null;
    try {
      if (this.lock.readLock().tryLock(maxTimeToLockInSec, TimeUnit.SECONDS)) {
        try {
          Iterator<SocketBasedHost> iterator = this.pool.keySet().iterator();
          while (iterator.hasNext()) {
            socketBasedHost = iterator.next();
            stat = getStats(socketBasedHost);
            sb.append(stat).append("\r\n");
          } 
        } finally {
          this.lock.readLock().unlock();
        } 
      } else {
        sb.append("N/A").append("\r\n");
      } 
    } catch (InterruptedException ex) {
      logger.log(Level.FINE, "InterruptedException{0}", ex);
      sb.append("N/A Er").append("\r\n");
    } 
    return sb.toString();
  }
  
  public String getStats(SocketBasedHost host) throws InterruptedException {
    ConcurrentLinkedQueue poolForHost = this.pool.get(host);
    ConcurrentLinkedQueue poolToHost = this.inUsePool.get(host);
    
    int freeSize = 0;
    int inuseSize = 0;
    
    if (this.lock.readLock().tryLock(maxTimeToLockInSec, TimeUnit.SECONDS)) {
      try {
        freeSize = poolForHost.size();
        inuseSize = poolToHost.size();
      } finally {
        this.lock.readLock().unlock();
      } 
    }
    
    StringBuilder sb = new StringBuilder();
    sb.append(host).append(", ");
    sb.append("FreeSize, ").append(freeSize).append(", ");
    sb.append("InUseSize, ").append(inuseSize);
    return sb.toString();
  }
  
  public void close() {
    if (this.noopThread != null) {
      this.noopThread.interrupt();
      this.noopThread = null;
    } 
    
    SocketBasedHost socketBasedHost = null;
    this.lock.writeLock().lock();
    try {
      Iterator<SocketBasedHost> iterator = this.pool.keySet().iterator();
      while (iterator.hasNext()) {
        socketBasedHost = iterator.next();
        cleanPool(socketBasedHost);
      } 
      
      this.pool.clear();
      this.inUsePool.clear();
    } finally {
      this.lock.writeLock().unlock();
    } 
  }
  
  public void sendNoOp() {
    SocketBasedHost socketBasedHost = null;
    ConcurrentLinkedQueue<PooledBlockingClient> mypool = null;
    
    Iterator<SocketBasedHost> iterator = this.pool.keySet().iterator();
    while (iterator.hasNext()) {
      socketBasedHost = iterator.next();
      mypool = this.pool.get(socketBasedHost);
      
      if (socketBasedHost.getStatus() == 'A') {
        if (isDebug()) logger.log(Level.FINEST, "Starting NOOP {0}; size: {1}", 
              new Object[] { socketBasedHost, Integer.valueOf(mypool.size()) }); 
        sendNoOp(socketBasedHost);
        if (isDebug()) logger.log(Level.FINEST, "Done NOOP {0}; size: {1}", 
              new Object[] { socketBasedHost, Integer.valueOf(mypool.size()) }); 
      } else {
        int size = getPoolSize(socketBasedHost);
        if (size > 0)
        { logger.log(Level.FINEST, "Host is not UP {0}; size: {1}", 
              new Object[] { socketBasedHost, Integer.valueOf(size) });
          cleanPool(socketBasedHost);
          size = getPoolSize(socketBasedHost);
          logger.log(Level.FINEST, "Done {0}; size: {1}", 
              new Object[] { socketBasedHost, Integer.valueOf(size) }); }
        
        else if (isDebug()) { logger.log(Level.FINEST, "Host is not UP {0}", 
              new Object[] { socketBasedHost }); }
      
      } 
      
      if (socketBasedHost.getStatus() == 'A') {
        int size = getPoolSize(socketBasedHost);
        if (size != -1) {
          if (size > this.idlePoolSize) {
            reduceSize(socketBasedHost, mypool);
          } else if (size < this.minPoolSize) {
            increaseSize(socketBasedHost, mypool);
          } 
        }
      } 
      
      checkForLeak(socketBasedHost);
    } 
  }
  
  private void reduceSize(SocketBasedHost host, ConcurrentLinkedQueue<PooledBlockingClient> poolForHost) {
    PooledBlockingClient pooledBlockingClient = null;
    
    int size = getPoolSize(host);
    logger.log(Level.FINEST, 
        "Start: Pool {0}; size is more then ideal size {1}; free size: {2}; fullsize: {3}", 
        new Object[] { host, Integer.valueOf(this.idlePoolSize), Integer.valueOf(poolForHost.size()), Integer.valueOf(size) });
    
    if (size == -1)
      return; 
    for (int i = 0; size > this.idlePoolSize && i < this.maxPoolSize; i++) {
      pooledBlockingClient = poolForHost.poll();
      if (pooledBlockingClient == null)
        break;  try {
        pooledBlockingClient.getBlockingClient().close();
      } catch (IOException ex) {
        Logger.getLogger(BlockingClientPool.class.getName()).log(
            Level.SEVERE, "Error closing: " + ex, ex);
      } 
      
      returnBlockingClient(pooledBlockingClient);
      pooledBlockingClient = null;
      
      size = getPoolSize(host);
      if (size == -1)
        break; 
    } 
    logger.log(Level.FINEST, 
        "End: Pool {0}; size was more then ideal size {1}; free size: {2}; fullsize: {3}", 
        new Object[] { host, Integer.valueOf(this.idlePoolSize), Integer.valueOf(poolForHost.size()), Integer.valueOf(size) });
  }
  
  private void increaseSize(SocketBasedHost host, ConcurrentLinkedQueue<PooledBlockingClient> poolForHost) {
    PooledBlockingClient pooledBlockingClient = null;
    
    int size = getPoolSize(host);
    logger.log(Level.FINEST, 
        "Start: Pool {0}; size is less then min size {1}; free size: {2}; fullsize: {3}", 
        new Object[] { host, Integer.valueOf(this.minPoolSize), Integer.valueOf(poolForHost.size()), Integer.valueOf(size) });
    if (size == -1)
      return; 
    for (int i = 0; size < this.minPoolSize && i < this.maxPoolSize; i++) {
      pooledBlockingClient = getNewPooledBlockingClient(host);
      if (pooledBlockingClient == null) {
        break;
      }
      poolForHost.add(pooledBlockingClient);
      
      size = getPoolSize(host);
      if (size == -1)
        break; 
    } 
    logger.log(Level.FINEST, 
        "End: Pool {0}; size was less then min size {1}; free size: {2}; fullsize: {3}", 
        new Object[] { host, Integer.valueOf(this.minPoolSize), Integer.valueOf(poolForHost.size()), Integer.valueOf(size) });
  }
  
  private void checkForLeak(SocketBasedHost host) {
    ConcurrentLinkedQueue poolForInUseHost = this.inUsePool.get(host);
    PooledBlockingClient pooledBlockingClient = null;
    
    if (this.inUsePool.isEmpty())
      return; 
    Iterator<PooledBlockingClient> iterator = poolForInUseHost.iterator();
    long timedef = 0L;
    List<PooledBlockingClient> listToRemove = new ArrayList();
    while (iterator.hasNext()) {
      pooledBlockingClient = iterator.next();
      if (pooledBlockingClient == null)
        continue;  timedef = System.currentTimeMillis() - pooledBlockingClient.getLastActionTime();
      if (timedef < 1000L)
        continue; 
      timedef /= 1000L;
      
      if (timedef > getPoolableBlockingClient().getMaxIntervalForBorrowInSec()) {
        logger.log(Level.WARNING, "There looks to be a leak {0}.. closing", pooledBlockingClient);
        listToRemove.add(pooledBlockingClient);
        pooledBlockingClient.close();
      } 
    } 
    if (!listToRemove.isEmpty()) {
      logger.log(Level.WARNING, "Total number of leaks {0}", Integer.valueOf(listToRemove.size()));
      try {
        if (this.lock.writeLock().tryLock(maxTimeToLockInSec, TimeUnit.SECONDS)) {
          try {
            poolForInUseHost.removeAll(listToRemove);
          } finally {
            this.lock.writeLock().unlock();
          } 
        } else {
          logger.fine("unbale to remove leaks.. will try next time..");
        } 
      } catch (InterruptedException ex) {
        logger.log(Level.FINE, "unbale to remove leaks.. will try next time..", ex);
      } 
    } 
  }
  
  private void sendNoOp(SocketBasedHost host) {
    ConcurrentLinkedQueue<PooledBlockingClient> poolForHost = this.pool.get(host);
    ConcurrentLinkedQueue<PooledBlockingClient> poolForInUseHost = this.inUsePool.get(host);
    if (poolForHost == null || poolForInUseHost == null) {
      throw new IllegalStateException("pool for host was null!");
    }
    
    int size = poolForHost.size();
    
    PooledBlockingClient pooledBlockingClient = null;
    for (int i = 0; i < size; i++) {
      this.lock.writeLock().lock();
      
      try { pooledBlockingClient = poolForHost.poll();
        if (pooledBlockingClient != null)
        { pooledBlockingClient.setHandedOut(true);
          pooledBlockingClient.setPoolToReturn(poolForHost);

          
          poolForInUseHost.add(pooledBlockingClient);
           }
        
        else
        
        { this.lock.writeLock().unlock(); break; }  } finally { this.lock.writeLock().unlock(); }

      
      if (pooledBlockingClient != null) {
        returnBlockingClient(pooledBlockingClient);
        pooledBlockingClient = null;
      } 
    } 
  }
  
  private boolean cleanPool(SocketBasedHost host) {
    ConcurrentLinkedQueue<PooledBlockingClient> poolForHost = this.pool.get(host);
    ConcurrentLinkedQueue poolForInUseHost = this.inUsePool.get(host);
    if (poolForHost == null || poolForInUseHost == null) {
      throw new IllegalStateException("pool for host was null! " + host);
    }
    
    Iterator<PooledBlockingClient> iterator = poolForInUseHost.iterator();
    PooledBlockingClient pooledBlockingClient = null;
    
    logger.log(Level.FINEST, "Start: Clean Pool {0};  {1}; free size: {2}; fullsize: {3}", 
        new Object[] { host, Integer.valueOf(this.minPoolSize), Integer.valueOf(poolForHost.size()), Integer.valueOf(poolForInUseHost.size()) });
    try {
      if (this.lock.writeLock().tryLock(maxTimeToLockInSec, TimeUnit.SECONDS)) {
        try {
          while (iterator.hasNext()) {
            pooledBlockingClient = iterator.next();
            if (pooledBlockingClient == null)
              continue;  pooledBlockingClient.setPoolToReturn(null);
            pooledBlockingClient.close();
          } 
          poolForInUseHost.clear();
          
          iterator = poolForHost.iterator();
          pooledBlockingClient = null;
          while (iterator.hasNext()) {
            pooledBlockingClient = iterator.next();
            if (pooledBlockingClient == null)
              continue;  pooledBlockingClient.close();
          } 
          poolForHost.clear();
        } finally {
          this.lock.writeLock().unlock();
        } 
        
        logger.log(Level.FINEST, "End: Clean Pool {0};  {1}; free size: {2}; fullsize: {3}", 
            new Object[] { host, Integer.valueOf(this.minPoolSize), Integer.valueOf(poolForHost.size()), Integer.valueOf(poolForInUseHost.size()) });
        return true;
      } 
      return false;
    }
    catch (InterruptedException ex) {
      logger.log(Level.FINE, "Error: {0}", ex);
      return false;
    } 
  }

  
  public PoolableBlockingClient getPoolableBlockingClient() { return this.poolableBlockingClient; }


  
  public void setPoolableBlockingClient(PoolableBlockingClient poolableBlockingClient) { this.poolableBlockingClient = poolableBlockingClient; }


  
  public String getName() { return this.name; }


  
  public void setName(String name) { this.name = name; }


  
  public int getMinPoolSize() { return this.minPoolSize; }


  
  public void setMinPoolSize(int minPoolSize) { this.minPoolSize = minPoolSize; }


  
  public int getMaxPoolSize() { return this.maxPoolSize; }


  
  public void setMaxPoolSize(int maxPoolSize) { this.maxPoolSize = maxPoolSize; }


  
  public int getIdlePoolSize() { return this.idlePoolSize; }


  
  public void setIdlePoolSize(int idlePoolSize) { this.idlePoolSize = idlePoolSize; }


  
  public HostMonitoringService getHostMonitoringService() { return this.hostMonitoringService; }


  
  private void setHostMonitoringService(HostMonitoringService hostMonitoringService) { this.hostMonitoringService = hostMonitoringService; }


  
  public int getLogPoolStatsTimeInMinute() { return this.logPoolStatsTimeInMinute; }


  
  public void setLogPoolStatsTimeInMinute(int logPoolStatsTimeInMinute) { this.logPoolStatsTimeInMinute = logPoolStatsTimeInMinute; }


  
  public String toStirng() { return getStats(); }
}
